#!/usr/bin/env bash

# Copyright Materialize, Inc. and contributors. All rights reserved.
#
# Use of this software is governed by the Business Source License
# included in the LICENSE file at the root of this repository.
#
# As of the Change Date specified in that file, in accordance with
# the Business Source License, use of this software will be governed
# by the Apache License, Version 2.0.

# Test that a catalog created by materialized v0.1.0 is compatible with the
# version of materialized under test.

set -euo pipefail

say() {
    echo "### $* ###"
}

run_sql() {
    psql -U materialize -h localhost -p 6875 materialize -c "\pset footer off" -c "$1" "$@"
}

launch_materialized() {
    name=$1
    shift
    "materialized-$name" "$@" &
    materialized_pid=$?
    wait-for-it --timeout=30 -q localhost:6875
    run_sql "SELECT 1" > /dev/null
}

kill_materialized() {
  kill "$materialized_pid"
  wait 2> /dev/null
}

wait-for-it --timeout=30 kafka:9092

say "launching materialized-golden010"
launch_materialized golden010

say "building golden010 catalog"
testdrive --kafka-addr kafka:9092 <<'EOF'
$ set schema={
    "type": "record",
    "name": "envelope",
    "fields": [
      {
        "name": "before",
        "type": [
          {
            "name": "row",
            "type": "record",
            "fields": [
              {"name": "a", "type": "long"},
              {"name": "b", "type": "long"}
            ]
          },
          "null"
        ]
      },
      { "name": "after", "type": ["row", "null"] }
    ]
  }

$ kafka-create-topic topic=real-time

$ kafka-ingest format=avro topic=real-time schema=${schema} timestamp=1
{"before": null, "after": {"row": {"a": 1, "b": 1}}}
{"before": null, "after": {"row": {"a": 2, "b": 1}}}
{"before": null, "after": {"row": {"a": 3, "b": 1}}}
{"before": null, "after": {"row": {"a": 1, "b": 2}}}

> CREATE SOURCE real_time_src
  FROM KAFKA BROKER '${testdrive.kafka-addr}' TOPIC 'testdrive-real-time-${testdrive.seed}'
  FORMAT AVRO USING SCHEMA '${schema}'
  ENVELOPE DEBEZIUM

> CREATE MATERIALIZED VIEW real_time AS
  SELECT *, concat(a::text, CAST(b AS text)) AS c
  FROM real_time_src

> CREATE INDEX real_time_idx ON real_time (a::text)

> SELECT * FROM real_time
a  b  c
--------
1  1  11
2  1  21
3  1  31
1  2  12

# Checks for regressions against #5802; simply having this in the catalog is
# sufficient, so it doesn't need to be requeried in each new "epoch".
> CREATE MATERIALIZED VIEW regression_5802 AS SELECT length(length('')::text);

> SELECT * FROM regression_5802
length
------
1
EOF

say "killing materialized-golden010"
kill_materialized

say "launching materialized-golden060"
launch_materialized golden060

say "adding golden060 features to golden010 catalog"
testdrive --no-reset <<'EOF'
> CREATE TABLE t (a int, b text DEFAULT 'def')
> INSERT INTO t (a) VALUES (42)
> SELECT * FROM t
42  def
EOF

say "killing materialized-golden060"
kill_materialized

say "launching materialized-golden071"
launch_materialized golden071

say "adding golden071 features to golden060 catalog"
testdrive --kafka-addr kafka:9092 --no-reset <<'EOF'
# Checks a view with a CTE.
> CREATE MATERIALIZED VIEW v AS WITH u AS (SELECT * FROM real_time_src) SELECT * from u;
EOF

say "creating Kafka source with metrics disabled"
testdrive --kafka-addr kafka:9092 --no-reset <<'EOF'
# Copy schema from above, golden010 catalog doesn't support statistics_interval_ms
$ set schema={
    "type": "record",
    "name": "envelope",
    "fields": [
      {
        "name": "before",
        "type": [
          {
            "name": "row",
            "type": "record",
            "fields": [
              {"name": "a", "type": "long"},
              {"name": "b", "type": "long"}
            ]
          },
          "null"
        ]
      },
      { "name": "after", "type": ["row", "null"] }
    ]
  }

> CREATE SOURCE real_time_src_no_stats
  FROM KAFKA BROKER '${testdrive.kafka-addr}' TOPIC 'testdrive-real-time-${testdrive.seed}'
  WITH (statistics_interval_ms = 0)
  FORMAT AVRO USING SCHEMA '${schema}'
  ENVELOPE DEBEZIUM

> CREATE MATERIALIZED VIEW real_time_no_stats AS
  SELECT *, concat(a::text, CAST(b AS text)) AS c
  FROM real_time_src_no_stats
EOF

say "adding table with char column"
testdrive --kafka-addr kafka:9092 --no-reset <<'EOF'
# Prior to v0.9.1 char references essentially aliased `text`, despite being its
# own entry in the catalog. The typmod value was also ignored.
> CREATE TABLE str_cols (a pg_catalog.char, b pg_catalog.char(1))

> SHOW CREATE TABLE str_cols
Table "Create Table"
--------------------
materialize.public.str_cols "CREATE TABLE \"materialize\".\"public\".\"str_cols\" (\"a\" \"pg_catalog\".\"char\", \"b\" \"pg_catalog\".\"char\"(1))"

> INSERT INTO str_cols VALUES ('abc', 'de'), ('uv', 'wxyz');

> SELECT a, b FROM str_cols ORDER BY a;
a b
--------
abc de
uv  wxyz
EOF

say "killing materialized-golden071"
kill_materialized

say "launching materialized-edge with golden catalog"
launch_materialized edge --log-file=stderr

# We use the final migration test twice, so any CREATE statement should use the
# IF NOT EXISTS version.
final_test=$(cat <<EOF
# NOTE(benesch): if/when tables persist their input data, we won't need to
# reinsert data here.
> INSERT INTO t (a) VALUES (42)
> INSERT INTO str_cols VALUES ('abc', 'de'), ('uv', 'wxyz');

> SELECT * FROM real_time
a  b  c
--------
1  1  11
2  1  21
3  1  31
1  2  12

> SELECT * FROM t
42  def

> SELECT * FROM v
a  b
----
1  1
2  1
3  1
1  2

# Kafka metrics for the real_time_src should be enabled now
# Count should be 2 because there are two materialized views on real_time_src
# If real_time_src_no_stats were also emitting stats, there would be 3 rows
> SELECT count(*) FROM mz_kafka_source_statistics;
count
-----
2

# Check that str_cols successfully migrated to a text type
> SHOW CREATE TABLE str_cols
Table "Create Table"
--------------------
materialize.public.str_cols "CREATE TABLE \"materialize\".\"public\".\"str_cols\" (\"a\" \"pg_catalog\".\"text\", \"b\" \"pg_catalog\".\"text\")"

> SELECT a, b FROM str_cols ORDER BY a;
a b
--------
abc de
uv  wxyz

# Check that char w/o typmod now behaves as char(1) in Postgres
> CREATE TABLE IF NOT EXISTS char_col (a char)
> SHOW CREATE TABLE char_col
Table "Create Table"
--------------------
materialize.public.char_col "CREATE TABLE \"materialize\".\"public\".\"char_col\" (\"a\" \"pg_catalog\".\"bpchar\")"

> INSERT INTO char_col VALUES ('abc'::char), ('wxyz'::char);

> SELECT a FROM char_col ORDER BY a;
a
----
a
w

EOF
)

say "validating edge state"
testdrive --kafka-addr kafka:9092 --no-reset <<<"$final_test"

say "killing materialized-edge"
kill_materialized

say "launching materialized-edge with golden catalog"
launch_materialized edge --log-file=stderr

say "validating edge state again to ensure migration idempotency"
testdrive --kafka-addr kafka:9092 --no-reset <<<"$final_test"
